package com.aispeech.asr.apis.ssc.beans;

import java.io.File;
import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Function;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Scope;
import org.springframework.core.io.FileSystemResource;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.client.MultipartBodyBuilder;
import org.springframework.stereotype.Component;
import org.springframework.util.MultiValueMap;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.WebClient;

import com.aispeech.asr.apis.ssc.config.AppConfig;
import com.aispeech.asr.apis.ssc.constants.Progress;
import com.aispeech.asr.apis.ssc.data.domain.Align;
import com.aispeech.asr.apis.ssc.data.domain.AudioInfo;
import com.aispeech.asr.apis.ssc.data.domain.TaskDoc;
import com.aispeech.asr.apis.ssc.data.domain.VprAudioInfo;
import com.aispeech.asr.apis.ssc.event.SscEvent;
import com.aispeech.asr.apis.ssc.service.ProgressManager;
import com.aispeech.asr.comm.data.form.vpr.VprApp;
import com.aispeech.asr.comm.data.form.vpr.VprAudio;
import com.aispeech.asr.comm.data.form.vpr.VprRequest;
import com.aispeech.asr.comm.enums.APICode;
import com.aispeech.asr.comm.enums.EAudioType;
import com.aispeech.asr.comm.enums.ErrorCode;
import com.aispeech.asr.comm.exceptions.AsrException;
import com.fasterxml.jackson.databind.JsonNode;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

@Slf4j
@Component
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public class Vprer implements Function<SscEvent, Mono<SscEvent>> {
    @Autowired
    WebClient webClient;
    @Autowired
    AppConfig config;
    @Autowired
    ProgressManager progressManager;

    @SneakyThrows
    @Override
    public Mono<SscEvent> apply(SscEvent event) {
    	TaskDoc taskDoc = event.getTaskDoc();
    	log.info("[ Vprer ] start。。。taskId:{}",taskDoc.getId());
        if (taskDoc.getProgress() >= Progress.VPRED) return Mono.just(event);
        taskDoc.addMetric("vpring_t");
        
        List<VprAudioInfo> prepareAudioInfos = taskDoc.getPrepareAudioInfos();
        log.info("[ Vprer ]: {}, total: {}", event.getTaskDoc().getId(), prepareAudioInfos.size());

        progressManager.syncToMongo(taskDoc, Progress.VPRING);
//        final Semaphore semaphore = new Semaphore(config.getNAsrParallels() - 2);

        return Flux.fromIterable(prepareAudioInfos)
                .publishOn(Schedulers.elastic())
                .flatMapSequential(prepareAudioInfo -> doVpr(event, prepareAudioInfo), config.getNVprParallels())
                .collectList()
                .map(asrContexts -> { //这样写可以吗？
                	//taskDoc.setPrepareAudioInfos(prepareAudioInfos);
                	updateResult(event,prepareAudioInfos);  
                	progressManager.syncToMongo(taskDoc, Progress.VPRED);
                	taskDoc.addMetric("vpred_t");
                	log.info("[ Vprer ] end。。。taskId:{}",taskDoc.getId());
                    return event;
                })
                .defaultIfEmpty(event);
    }
    
    private void updateResult(SscEvent event,List<VprAudioInfo> prepareAudioInfos) {

    	TaskDoc taskDoc = event.getTaskDoc();
    	Collections.sort(prepareAudioInfos,new Comparator<VprAudioInfo>(){
            public int compare(VprAudioInfo arg0, VprAudioInfo arg1) {
            	float flag = arg0.getScore() - arg1.getScore();
            	if(flag > 0) {
            		return -1;
            	}else if(flag < 0) {
            		return 1;
            	}
                return 0;
            }
        });
    	String staffId = prepareAudioInfos.get(0).getUid(); //从大到小
    	
    	log.info("vpr collectList updateResult taskId:{}, staffId():{}",taskDoc.getId(), staffId);
    	
    	List<Map<String, Object>> sscVprMapList = new ArrayList<Map<String, Object>>();
    	for(Align align: taskDoc.getAligns()) {
    		String uid = String.valueOf(align.getUid());
    		int staff = staffId.equals(uid) ? 1:0;
    		Map<String, Object> map = new HashMap<String,Object>();
    		map.put("bg", align.getStart());
    		map.put("ed", align.getEnd());
    		map.put("speaker_ssc", uid);
    		map.put("staff", staff);
    		sscVprMapList.add(map);
    	}
    	taskDoc.setResult(sscVprMapList);
    	
    	log.info("vpr collectList updateResult taskId:{},sscVprMapList.size():{}",taskDoc.getId(), sscVprMapList.size());
    }

    @SneakyThrows
    public Mono<SscEvent> doVpr(SscEvent event, VprAudioInfo prepareAudioInfo) {
        String filePath = prepareAudioInfo.getAudioPath();
        File audio = new File(filePath);

        if (audio.exists()) {
            return Mono.just(event)
            		.flatMap(ctx -> vprStart(ctx, prepareAudioInfo))
            		.flatMap(session -> vprVerify(event, prepareAudioInfo, session));
        } else {
            return Mono.error(new AsrException(ErrorCode.ERR_ASR_FAILED.errid,
                    String.format("audio: %s not exists.", filePath)
            ));
        }
    }
    
    
    private Mono<String> vprStart(SscEvent event,VprAudioInfo prepareAudioInfo) {
    	
    	TaskDoc taskDoc = event.getTaskDoc();
        String filePath = prepareAudioInfo.getAudioPath();
        File audio = new File(filePath);
        
        AudioInfo audioInfo = event.getAudioInfo();
        int opcode = audioInfo.getSampleRate() == 8000 ? 1:4;

        if (audio.exists()) {
            VprRequest param = VprRequest.of(
            		VprAudio.of(EAudioType.WAV.key(),audioInfo.getSampleRate()),
            		Arrays.asList(new Integer[]{Integer.valueOf(taskDoc.getRequest().get("userId").toString())}),
            		taskDoc.getRequest().get("organization").toString(),
            		VprApp.of(taskDoc.getProductId(),"vpr_scc")
            );

            String vprHost =  config.getVprUrl();
            log.info("do VPR taskId: {}, uid: {}, url: {}, param: {}", event.getTaskDoc().getId(), prepareAudioInfo.getUid(),
            		vprHost, param);

            return webClient.post()
                    .uri(URI.create(String.format("%s/v1/start/verify?opcode=%d&from=asr_ssc&productId=%s", vprHost,opcode, event.getTaskDoc().getProductId())))
                    //.body(BodyInserters.fromMultipartData(parts))
                    .contentType(MediaType.APPLICATION_JSON)
                    .bodyValue(param)
                    .retrieve()
                    .onStatus(HttpStatus::is4xxClientError, this::createException)
                    .onStatus(HttpStatus::is5xxServerError, this::createException)
                    .bodyToMono(JsonNode.class)
                    .retry(3)
                    .map(jsonNode -> processPrepareInfo(event, prepareAudioInfo, jsonNode));
        } else {
            return Mono.error(new AsrException(ErrorCode.ERR_ASR_FAILED.errid,
                    String.format("audio: %s not exists.", filePath)
            ));
        }
    }
    
    @SneakyThrows
    private String processPrepareInfo(SscEvent sscEvent, VprAudioInfo prepareAudioInfo, JsonNode jsonNode) {
        log.info("vpr PrepareInfo taskId:{}, res: {}, uid: {}",sscEvent.getTaskDoc().getId(), jsonNode, prepareAudioInfo.getUid());
        String session = "";
        int state = jsonNode.get("state").intValue();
        if(state == 0) {
        	session = jsonNode.get("data").get("session").textValue();
        }else {
        	throw new AsrException(ErrorCode.ERR_VPR_FAILED.errid, ErrorCode.ERR_VPR_FAILED.error);
        }
        log.info("vpr PrepareInfo end taskId:{},session: {}, uid: {}",sscEvent.getTaskDoc().getId(), session, prepareAudioInfo.getUid());
        return session;
    }
    
    
    private Mono<SscEvent> vprVerify(SscEvent event,VprAudioInfo prepareAudioInfo,String session) {
    	TaskDoc taskDoc = event.getTaskDoc();
        String filePath = prepareAudioInfo.getAudioPath();
        File audio = new File(filePath);
        
        AudioInfo audioInfo = event.getAudioInfo();
        int opcode = audioInfo.getSampleRate() == 8000 ? 1:4;
        
        if(StringUtils.isBlank(session)) {
    		return Mono.error(new AsrException(ErrorCode.ERR_ASR_FAILED.errid,
                    String.format("audio: %s not exists.", filePath)
            ));
    	}

        if (audio.exists()) {
            MultipartBodyBuilder builder = new MultipartBodyBuilder();
            builder.part("file", new FileSystemResource(filePath));
            
            MultiValueMap<String, HttpEntity<?>> parts = builder.build();

            String vprHost =  config.getVprUrl();
            log.info("do VPR: {}, uid: {}, url: {}", taskDoc.getId(), prepareAudioInfo.getUid(),
            		vprHost);

            return webClient.post()
                    .uri(URI.create(String.format("%s/v1/verify?opcode=%d&session=%s&from=asr_ssc&productId=%s", vprHost,opcode,session, event.getTaskDoc().getProductId())))
                    .body(BodyInserters.fromMultipartData(parts))
                    .retrieve()
                    .onStatus(HttpStatus::is4xxClientError, this::createException)
                    .onStatus(HttpStatus::is5xxServerError, this::createException)
                    .bodyToMono(JsonNode.class)
                    .retry(3)
                    .map(jsonNode -> processVerify(event, prepareAudioInfo, jsonNode));
        } else {
            return Mono.error(new AsrException(ErrorCode.ERR_ASR_FAILED.errid,
                    String.format("audio: %s not exists.", filePath)
            ));
        }
        
    	//return Mono.just(event);
    }
    
    @SneakyThrows
    private SscEvent processVerify(SscEvent sscEvent, VprAudioInfo prepareAudioInfo, JsonNode jsonNode) {
    	TaskDoc taskDoc = sscEvent.getTaskDoc();
        log.info("vpr verify taskId:{},jsonNode: {}, uid: {}",taskDoc.getId(), jsonNode, prepareAudioInfo.getUid());
        
        if(jsonNode.get("state") != null ) {
        	//int state = jsonNode.get("state").intValue();
        	try {
        		if(jsonNode.get("data").get("verify_result").get("models").get(0).get("score") != null) {
            		float score = jsonNode.get("data").get("verify_result").get("models").get(0).get("score").floatValue();
                	log.info("vpr taskId:{},uid: {},score:{}",taskDoc.getId(), prepareAudioInfo.getUid(),score);
                	prepareAudioInfo.setScore(score);
                	prepareAudioInfo.setVprResult(jsonNode.textValue());
            	}
        	}catch (Exception e) {
        		float score = Float.MIN_VALUE;
            	prepareAudioInfo.setScore(score);
            	prepareAudioInfo.setVprResult(jsonNode.textValue());
			}
        }
        return sscEvent;
    }

    private Mono<AsrException> createException(ClientResponse response) {
        return response
                .createException()
                .map(ex -> new AsrException(APICode.INTERNAL_ERROR.getCode(), ex.getResponseBodyAsString()));
    }

   
}
